package com.intct.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import com.intct.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gufg
 * @since 2025-07-29 16:09
 */
public class DwdVehicleApp {
    public static void main(String[] args) throws Exception {
        // 1 基本环境配置
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.1 设置并行度
        env.setParallelism(1);

        // 1.2 启动检查
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 2获取源数据  [模拟 --> 日志文件 --> flume --> ods(kafka)]
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(Constant.KAFKA_ODS_VEHICLE_TOPIC, "dwd-vehicle-group-id");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dwd-vehicle-source");

        // 3 处理源数据
        // 3.1 过滤数据(不在线车辆)
        SingleOutputStreamOperator<String> filterDS = sourceDS.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String jsonStr) throws Exception {
                // 转换为JSONObject
                JSONObject jsonObj = JSONObject.parseObject(jsonStr);

                JSONObject afterObj = jsonObj.getJSONObject(Constant.MYSQL_CDC_JSON_KEY_AFTER);

                // 车辆是否在线
                String type = afterObj.getString("type");

                // 在线车辆
                return "0".equals(type);
            }
        });

        /*
            样例：
            {
                "op": "r",
                "name": "vehicle_distribution",
                "after": {
                    "actionTime": 1753773834,       时间戳
                    "areaId": "110119",             区域ID
                    "driverId": "870",              司机ID
                    "carShelfNumber": "Y51637129D", 车架号
                    "car": "24",                     车牌
                    "type": "0",                    车辆是否在线
                    "speed": 16                     速度
                },
                "ts_ms": 1753773834880
            }
         */
        // 将不需要的字段去掉
        SingleOutputStreamOperator<String> mapDS = filterDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String jsonStr) throws Exception {
                // 转换为JSONObject
                JSONObject jsonObj = JSONObject.parseObject(jsonStr);

                JSONObject afterObj = jsonObj.getJSONObject(Constant.MYSQL_CDC_JSON_KEY_AFTER);

                // TODO 4 规范
                // TODO 4.1 重复数据
                // TODO 4.2 空值
                // TODO 4.3 脱敏
                // TODO 4.4 单位

                return JSON.toJSONString(afterObj);
            }
        });

        // 5 写出数据(dwd --> kafka)
        KafkaSink<String> kafkaSink = KafkaUtil.getKafkaSink(Constant.KAFKA_DWD_VEHICLE_TOPIC, "dwd-trans-id-prefix");
        mapDS.sinkTo(kafkaSink);

        // 6 启动作业
        env.execute();
    }
}
